package com.hegaoye.mqhandler;

import com.alibaba.fastjson.JSON;
import com.hegaoye.base.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.Date;

@Slf4j
@Component
public class KafkaSender {

  @Resource private KafkaTemplate<String, String> kafkaTemplate;

  public void send(String msg) {
    String topic = "test";
    Message message = new Message();
    message.setMsg(msg);
    message.setSendTime(new Date());
    log.info("========发送消息>>>>{}<<<<<==========", JSON.toJSONString(message));
    // 发送
    ListenableFuture<SendResult<String, String>> future =
        kafkaTemplate.send(topic, JSON.toJSONString(message));
    // 添加成功发送消息的回调和失败的回调
    future.addCallback(
        result -> log.info("send message to {} success", topic),
        ex -> log.info("send message to {} failure,error message:{}", topic, ex.getMessage()));
  }

  /**
   * 发送topic 消息
   *
   * @param message
   */
  public void sendTopic(String message) {
    log.info("OrderSender >> " + message);
    kafkaTemplate.send(KafkaConfig.ORDER_DISCOUNT_QUEUE_NAME, message);
  }
}
